package com.sw.rpc.consumer;

import com.sw.rpc.service.JSONSerializer;
import com.sw.rpc.service.RpcDecoder;
import com.sw.rpc.service.RpcEncoder;
import com.sw.rpc.service.RpcRequest;
import com.sw.rpc.service.zk.RpcResponse;
import com.sw.rpc.service.zk.ServiceInstanceService;
import com.sw.rpc.service.zk.ZkNodeData;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.concurrent.Future;
import org.I0Itec.zkclient.IZkChildListener;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author sw
 * @project 连接zk
 * @description
 * @date 2020/5/15 13:38
 */
public class RpcConsumerZK {

    private static ExecutorService executor= Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    /**
     * 客户端信息
     */
    private static Map<String, UserClientHandler> userClientHandlers = new HashMap<>();
    private static Map<String, Channel> channels = new HashMap<>();
    private static  EventLoopGroup eventLoopGroup;
    static {
        //初始化服务连接 及服务在zk的节点监听
        try {
            initConnectServerInfo();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    /**
     * 创建代理类的执行信息
     * @param serviceClass
     * @param rpc
     * @return
     */
    public Object createProxy(final Class<?> serviceClass){
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),new Class[]{serviceClass},new InvocationHandler(){
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setRequestId(String.valueOf(UUID.randomUUID()));
                rpcRequest.setClassName(serviceClass.getName());
                rpcRequest.setMethodName(method.getName());
                rpcRequest.setParameterTypes(method.getParameterTypes());
                rpcRequest.setParameters(args);
                //发送消息
                return sendMessage();
            }

        });



    }
    public static Object sendMessage() throws ExecutionException, InterruptedException {
        if (channels.size() < 1) {
            return null;
        }
        String key = getFirstKeyOfMap(channels);
        Channel channel = channels.get(key);
        UserClientHandler userClientHandler = userClientHandlers.get(key);
        userClientHandler.setChannel(channel);
        // 设置参数
        userClientHandler.setPara(UUID.randomUUID().toString());
        // 去服务端请求数据
        Object result = executor.submit(userClientHandler).get();
        if (result.equals("timeout")) {
            // 消息超时，超时重发
            System.out.println("消息超时，重新发送到" + key);
            sendMessage();
        }
        return result;
    }


    public static void connectServer(ZkNodeData zkNodeData) throws InterruptedException {
        UserClientHandler userClientHandler = new UserClientHandler();
        if(null==eventLoopGroup){
            eventLoopGroup = new NioEventLoopGroup();
        }
        userClientHandlers.put(getZNodeKey(zkNodeData), userClientHandler);
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                       /* pipeline.addLast(new RpcDecoder(RpcResponse.class,new JSONSerializer()));
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                       */
                       pipeline.addLast(new StringDecoder());
                       pipeline.addLast(new StringEncoder());
                        pipeline.addLast(userClientHandler);
                    }
                });
        ChannelFuture connect = bootstrap.remoteAddress(zkNodeData.getIp(), zkNodeData.getPort())
                .connect();
        connect.sync();
        connect.addListener((ChannelFuture future)->{
            Channel channel = future.channel();
            final EventLoop eventLoop = channel.eventLoop();
            if (future.isSuccess()) {
                System.out.println(zkNodeData + "连接成功");
                channels.put(getZNodeKey(zkNodeData), channel);
            } else {
                System.out.println(zkNodeData.getPort() + "连接失败");
                future.cause().printStackTrace();
                eventLoop.schedule(() -> doConnect(bootstrap, zkNodeData), 10, TimeUnit.SECONDS);
            }
        });

    }

    private static String getZNodeKey(ZkNodeData zkNodeData) {
        return zkNodeData.getIp()+":"+zkNodeData.getPort();
    }

    /**
     * 执行服务端连接
     * @param bootstrap
     * @param zNode
     */
    private static void doConnect(Bootstrap bootstrap, ZkNodeData zNode) {
        try {
            if (bootstrap != null) {
                bootstrap.remoteAddress(zNode.getIp(), zNode.getPort());
                ChannelFuture future = bootstrap.connect().addListener((ChannelFuture futureListener) -> {
                    final EventLoop eventLoop = futureListener.channel().eventLoop();
                    if (!futureListener.isSuccess()) {
                        //连接tcp服务端不成功 10后重连
                        System.out.println(zNode.toString() + "与服务端 断开连接!在10s之后准备尝试重连!");
                        eventLoop.schedule(() -> doConnect(bootstrap, zNode), 10, TimeUnit.SECONDS);
                    }
                });
                channels.put(zNode.getIp() + ":" + zNode.getPort(), future.channel());
            }
        } catch (Exception e) {
            System.out.println("客户端连接失败!" + e.getMessage());
        }
    }

    /**
     * 连接服务
     */
    public static void initConnectServerInfo() throws InterruptedException {
        List<String> childrenServer = ServiceInstanceService.client.getChildren(ServiceInstanceService.SERVERINSTANCEURL);
        //存在数据，执行连接
        for(String server:childrenServer){
            ZkNodeData zkNodeData = ServiceInstanceService.getZKNodeData(server);
            connectServer(zkNodeData);
        }
        /**
         * 监听服务节点下的信息
         */
        ServiceInstanceService.client.subscribeChildChanges(ServiceInstanceService.SERVERINSTANCEURL, new IZkChildListener() {
            @Override
            public void handleChildChange(String s, List<String> list) throws Exception {
                //判断当前连接的是否存在，
                //1.本地连接存在，但是zk服务已经下线,并删除
                for(Map.Entry<String,Channel> conneted:channels.entrySet()){
                    if(!list.contains(conneted.getKey())){
                        disconnectServer(conneted.getKey());
                    }
                }
                //2.新增的建立连接
                for(String newServer:list){
                    //
                    if(!channels.containsKey(newServer)){
                        ZkNodeData zkNodeData = ServiceInstanceService.getZKNodeData(newServer);
                        connectServer(zkNodeData);
                    }
                }

            }
        });
    }

    public static void disconnectServer(String url) {
        if (!channels.containsKey(url)) {
            System.out.println("未连接该服务端");
            return;
        }
        System.out.println("关闭与服务端" + url + "的连接");
        Channel channel = channels.get(url);
        Future<?> shutdownFuture = channel.disconnect();
        shutdownFuture.addListener((future) -> {
            if (future.isSuccess()) {
                userClientHandlers.remove(url);
                channels.remove(url);
                System.out.println(url + "关闭成功");
            } else {
                System.out.println(url + "关闭失败");
                future.cause().printStackTrace();
            }
        });
    }
    public static String getFirstKeyOfMap(Map<String, Channel> map) {
        return (String) map.keySet().toArray()[0];
    }
}
